package cn.finder.mq;

import java.io.UnsupportedEncodingException;

import javax.net.ssl.SSLSocketFactory;

import org.apache.log4j.Logger;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttClientPersistence;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.MqttSecurityException;

import cn.finder.FinderSSL;

/**
 *
 * @author finder
 *
 * suggest: topic rules
 *  // msgall  -> all client:   msgall/#
	// msggroup->point a group topic:  msggroup/{groupid}/#
	// msgone  -> send to single client:  msgone/clientid/#
	 //if  client_id is same, not allow subscribe for the thecond subscriber
	  // one more protocal call send or receive at the same time， you can disable tcp
 */
public class FinderMqttClient extends MqttClient{

	

	private static Logger logger = Logger.getLogger(FinderMqttClient.class);
	
	public final static int QOS_0=0;
	public final static int QOS_1=1;
	public final static int QOS_2=2;
	
	protected int qos = QOS_2; // service quaity
	private String userName = "mqtt";
	private String password = "mqtt";


	private int connectionTimeout = 10;
	private int keepAliveInterval = 20;

	private String[] brokeUrls;


	protected boolean retained = true;
	
	private boolean cleanSession=false;
	
	
	
	// disconnect then  reconnect timer ,unit is milsecond, default value is 1 minutes
	private long reconnectTimer=1000;
	
	
	// at starter, if can't connect , is need to reconnect, default is false
	private boolean reconnectWithStart=false;
	
	
	//is current state , record is try connect now, avoid open more thread
	private boolean isTryConnecting=false;
	
	private static Object lock=new Object();
	
	private static Object reconnectLock=new Object();
	
	MqttConnectOptions connOpts = new MqttConnectOptions();
	
	private FinderMqttCallback mqttCallback;
	
	

	public FinderMqttClient(String clientId,String serverURI) throws MqttException {
		super(serverURI, clientId);
		
	}
	
	public FinderMqttClient(String clientId,String serverURI,  MqttClientPersistence persistence) throws MqttException {
		super(serverURI, clientId, persistence);
	}

	
	/***
	 * get MqttClient interface
	 * 
	 * @return
	 * @throws MqttException
	 */
	public void initConnect() {

		if (!isConnected()) {
			synchronized (lock) {

				if (!isConnected()) {
					try {
						connectEx();
					} catch (MqttException e) {
						logger.error(e);
					}
				}
			}
		}
	}

	
	private void connectEx() throws MqttSecurityException, MqttException{
		
		connOpts.setCleanSession(cleanSession); // 生产者会话保持
		connOpts.setUserName(userName);
		connOpts.setPassword(password.toCharArray());
		connOpts.setConnectionTimeout(connectionTimeout);
		connOpts.setKeepAliveInterval(keepAliveInterval);
		connOpts.setAutomaticReconnect(reconnectWithStart);
		if (brokeUrls != null) {
			// String[] uris =
			// {"tcp://10.100.124.206:1883","tcp://10.100.124.207:1883"};
			connOpts.setServerURIs(brokeUrls); // for load balance
		}
		
		try{
			connect(connOpts);
		}
		catch(Exception e){
			logger.error(e);
		}
		finally{
			
			if(reconnectWithStart){
				if(!isConnected())
					reconnect();
			}
			
		}
		
	}
	

	
	
	/***
	 * publish
	 * publisher can not be sure the msg is reached to receiver
	 * but it can be sure the msg is be secure and sent to broker
	 * @param msg
	 * @param topic
	 * @throws MqttException
	 * @throws MqttPersistenceException
	 * @throws UnsupportedEncodingException 
	 */
	public void publishEx(String msg, String topic) throws MqttPersistenceException, MqttException, UnsupportedEncodingException {

		MqttMessage message = new MqttMessage(msg.getBytes("UTF-8"));
		message.setQos(qos);
		message.setRetained(retained); // set send save msg
		publish(topic, message);

	}


	
	public String getUserName() {
		return userName;
	}

	public void setUserName(String userName) {
		this.userName = userName;
	}

	public String getPassword() {
		return password;
	}

	public void setPassword(String password) {
		this.password = password;
	}

	public int getQos() {
		return qos;
	}

	public void setQos(int qos) {
		this.qos = qos;
	}



	public boolean isCleanSession() {
		return cleanSession;
	}

	public void setCleanSession(boolean cleanSession) {
		this.cleanSession = cleanSession;
	}

	
	@Override
	public void setCallback(MqttCallback mqttCallback){
		this.mqttCallback = (FinderMqttCallback)mqttCallback;
		aClient.setCallback(mqttCallback);
	}
	
	public void setReconnectTimer(long reconnectTimer){
		this.reconnectTimer=reconnectTimer;
	}
	
	
	

	public boolean isReconnectWithStart() {
		return reconnectWithStart;
	}

	public void setReconnectWithStart(boolean reconnectWithStart) {
		this.reconnectWithStart = reconnectWithStart;
	}

	
	
	public MqttConnectOptions getConnOpts() {
		return connOpts;
	}
	
	
	/***
	 * enable ssl transfer
	 * instead by setSSLParam
	 * @param caPath
	 * @param crtPath
	 * @param keyPath
	 * @param password
	 * @throws Exception
	 */
	@Deprecated 
	public void userSSL(String caPath,String crtPath, String keyPath, String password) throws Exception{
		SSLSocketFactory socketFactory=new FinderSSL().createSSLSocktet(caPath, crtPath, keyPath, password);
		connOpts.setSocketFactory(socketFactory);
	}
	
	public void setSslParam(SslParam sslParam)throws Exception{
		SSLSocketFactory socketFactory=new FinderSSL().createSSLSocktet(sslParam.getCaPath(), sslParam.getCrtPath(), sslParam.getKeyPath(), sslParam.getPassword());
		connOpts.setSocketFactory(socketFactory);
	}
	

	/***
	 * override reconnect
	 */
	@Override
	public void reconnect(){
		
		if(!isTryConnecting){
			synchronized (reconnectLock) {
				
				if(!isTryConnecting){
					new Thread(new Runnable(){
						@Override
						public void run() {
							doReconnect();
						}
						
					}).start();
				}
			}
			
		}
	}
	
	private void doReconnect(){
		while(!isConnected()){
			try {
			   mqttCallback.tryConnecting();
			   connect(connOpts);
				
			   if(isConnected()){
				 //when disconnect ,we can resubscribe, so we will callback method to tell caller.
				   mqttCallback.connectComplete(true, this.getServerURI());
				   break;
			   }
			  
			   
			} catch (Throwable e) {
				logger.error(e);
			}
			try {
				Thread.sleep(reconnectTimer);
			} catch (InterruptedException e) {
				logger.error(e);
			}
			logger.info(String.format("try connect to broke [%s]", this.getCurrentServerURI()));
		}
		
		logger.info(String.format("connected to broke [%s]", this.getCurrentServerURI()));
		
	}

	
	
	

	/***
	 * the base class of CallBack 
	 * @author finder
	 *
	 */
	public abstract static class FinderMqttCallback implements MqttCallbackExtended{
		
		private FinderMqttClient mqttClient;
		
		public  FinderMqttCallback(FinderMqttClient mqttClient){
			this.mqttClient=mqttClient;
		}
		
		// this method. we suggest not to use framework work,
		// because framework is not have feedback for connect success, and connect time is more
		@Override
		public void connectionLost(Throwable cause) {
			if(!mqttClient.isConnected())
				mqttClient.reconnect();
		}

		public abstract void messageArrived(String topic, MqttMessage message) throws Exception;
		

		@Override
		public abstract void deliveryComplete(IMqttDeliveryToken token);

		/***
		 * try to reconnecting , we can do something at time, 
		 * for example. tip process  or status, 
		 * you can do this method
		 */
		public void tryConnecting(){
			
		}
		
		/***
		 * if reconnect, we can resubscribe accrodding this method
		 */
		@Override
		public void connectComplete(boolean reconnect, String serverURI) {
			
		}

	}
}
